package prictice.base;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.Map;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ConcurrentLinkedDeque;

/**
 * @Package prictice.base
 * @Description:
 * @Author elwyn
 * @Date 2017/9/15 22:26
 * @Email elonyong@163.com
 */
public class SourceSpout implements IRichSpout {
    Queue<String> queue = new ConcurrentLinkedDeque<>();
    SpoutOutputCollector spoutOutputCollector;
    String str;

    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.spoutOutputCollector = spoutOutputCollector;

        Random random = new Random();
        String[] host = {"www.taobao.com"};
        String[] sessionId = {"HJKHKJHUIFDISH21312321", "JKHIHNIDFSFDS34535", "FDSFDSFDSFDSMKMK908", "DFSFDSFDSAAAA1123245", "GFGDGFDGHKINNIU778"};
        String[] time = {"2014-01-07 08:40:50", "2014-01-07 08:40:50",
                "2014-02-07 06:40:50", "2014-01-27 06:40:50",
                "2014-03-07 09:40:50", "2014-01-17 11:40:50",
                "2014-04-07 08:40:50", "2014-11-08 12:40:50",
                "2014-05-07 08:40:50"};
        for (int i = 0; i < 100; i++) {
            queue.add(host[0] + "\t" + sessionId[random.nextInt(5)] + "\t" + time[random.nextInt(9)]);

        }
    }

    @Override
    public void close() {

    }

    @Override
    public void activate() {

    }

    @Override
    public void deactivate() {

    }

    @Override
    public void nextTuple() {
        if (queue.size() >= 0) {
            spoutOutputCollector.emit(new Values(queue.poll()));
        }
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void ack(Object o) {
        System.out.println("spout ack:" + o.toString());
    }

    @Override
    public void fail(Object o) {
        System.out.println("spout fail:" + o.toString());

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("log"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
